package com.pay.consumer;

import com.alibaba.fastjson.JSONObject;
import com.pay.bo.NotifyMerchantBO;
import com.pay.config.RabbitMQSpareTireConfig;
import com.pay.entity.ConsumerLog;
import com.pay.service.Impl.ConsumerLogServiceImpl;
import com.pay.service.PayMessageTemplateService;
import com.pay.utils.RedisUtils;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;

/**
 * @author zhouyihe
 * @ClassName WechatTemplateConsumer
 */
@Component
@Slf4j
@RabbitListener(queues = RabbitMQSpareTireConfig.WECHAT_BATCH_TEMPLATE_QUEUE)
public class WechatTemplateBatchConsumer {
    @Autowired
    private PayMessageTemplateService payMessageTemplateService;
    @Autowired
    private ConsumerLogServiceImpl consumerLogServiceImpl;
    
    @Value("${spring.rabbitmq.listener.simple.retry.max-attempts}")
    private Integer maxAttempts;
    
    @Autowired
    private AmqpTemplate amqpTemplate;
    
    /**
     * 消费者订阅微信模板队列主题  调用 微信公众号模板接口
     *
     * @param msg
     * @param message
     * @param channel
     */
    @RabbitHandler
    public void process(String msg, Message message, Channel channel) throws IOException, IOException {
        log.info(">>调用微信模板接口发送微信模板提醒:{}<<", msg);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 要先将一个msg拆分成多个msg ,直接利用JSONObject里的方式进行拆分
            List<NotifyMerchantBO> notifyMerchantBOS = JSONObject.parseArray(msg, NotifyMerchantBO.class);
            notifyMerchantBOS.forEach((notifyMerchantBO) -> {
                // 为了解决幂等性问题 提前根据消息id查询 消费者是否消费过
                String msgId = notifyMerchantBO.getPaymentId();
                ConsumerLog consumerLog = consumerLogServiceImpl.getByMsgIdConsumerLog(msgId);
                if (consumerLog != null) {
                    // 已经消费过 手动ack告诉我们的mq删除该msg 避免重复消费
                    return;
                }
                // 1.调用第三方微信公众号模板接口
                // 2.该方法返回值true 手动ack成功 返回false手动ack失败 需要MQ给我们消费者补偿
                boolean result = payMessageTemplateService.notifyMerchantPaymentResult(notifyMerchantBO);
                if (!result) {
                    // 返回false手动ack失败 需要MQ给我们消费者补偿
                    // 拒绝签收 抛出异常（mq服务器端间隔一段重试该msg）
                    log.error("[调用消息模板接口发送消息失败,notifyMerchantBO:{}]", notifyMerchantBO);
                    throw new RuntimeException("调用消息模板接口发送消息失败..");
                }
            });
        } catch (Exception e) {
            log.error("e:{}", e);
            // 记录重试次数 重试多次还是失败的情况下  将批量msg 写日志表 死信队列
            // 1.获取重试次数
            String msgCount = RedisUtils.getString(Constant.REDIS_PAY_MSG + msg);
            int count = msgCount == null ? 0 : Integer.parseInt(msgCount);
            
            // 2.计数
            count++;
            
            // 2.判断是否到达重试最大次数
            if (count >= maxAttempts){
                amqpTemplate.convertAndSend(RabbitMQSpareTireConfig.RETRY_EXCHANGE,"",msg);
                log.info("消费者消费该msg:{}失败{}次，将msg存放死信队列中", msg, count);
                // 清空redis
                RedisUtils.delKey(Constant.REDIS_PAY_MSG + msg);
                return;
            }
            
            // 将计数存入redis中
            RedisUtils.setString(Constant.REDIS_PAY_MSG + msg, count);
            throw new RuntimeException(e);
        }
    }
}